package com.ferry.transport.netty.consumer;

import com.ferry.transport.api.Connection;
import com.ferry.transport.api.UnResolvedAddress;
import com.ferry.transport.api.config.ChildConfig;
import com.ferry.transport.api.config.TcpConfig;
import com.ferry.transport.api.domain.Protocol;
import com.ferry.transport.netty.provider.SocketChannelProvider;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollMode;
import io.netty.channel.kqueue.KQueue;
import io.netty.channel.kqueue.KQueueEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

import java.util.concurrent.ThreadFactory;

/**
 * Created by daiyong
 */
public abstract class NettyTcpConnector extends NettyConnector {

	private ChildConfig childConfig = new ChildConfig();

	public NettyTcpConnector() {
		super(Protocol.TCP);
		init();
	}

	public NettyTcpConnector(Protocol protocol) {
		super(protocol);
		init();
	}

	public NettyTcpConnector(int workers) {
		super(Protocol.TCP, workers);
		init();
	}

	@Override
	protected void setOptions() {
		super.setOptions();

		Bootstrap boot = bootstrap();

		// child options
		ChildConfig child = childConfig;

		boot
				.option(ChannelOption.SO_REUSEADDR, child.isReuseAddress())
				.option(ChannelOption.SO_KEEPALIVE, child.isKeepAlive())
				.option(ChannelOption.TCP_NODELAY, child.isTcpNoDelay())
				.option(ChannelOption.ALLOW_HALF_CLOSURE, child.isAllowHalfClosure());
		if (child.getRcvBuf() > 0) {
			boot.option(ChannelOption.SO_RCVBUF, child.getRcvBuf());
		}
		if (child.getSndBuf() > 0) {
			boot.option(ChannelOption.SO_SNDBUF, child.getSndBuf());
		}
		if (child.getLinger() > 0) {
			boot.option(ChannelOption.SO_LINGER, child.getLinger());
		}
		if (child.getIpTos() > 0) {
			boot.option(ChannelOption.IP_TOS, child.getIpTos());
		}
		if (child.getConnectTimeoutMillis() > 0) {
			boot.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, child.getConnectTimeoutMillis());
		}
		if (child.getTcpNotSentLowAt() > 0) {
			boot.option(EpollChannelOption.TCP_NOTSENT_LOWAT, child.getTcpNotSentLowAt());
		}
		if (child.getTcpKeepCnt() > 0) {
			boot.option(EpollChannelOption.TCP_KEEPCNT, child.getTcpKeepCnt());
		}
		if (child.getTcpUserTimeout() > 0) {
			boot.option(EpollChannelOption.TCP_USER_TIMEOUT, child.getTcpUserTimeout());
		}
		if (child.getTcpKeepIdle() > 0) {
			boot.option(EpollChannelOption.TCP_KEEPIDLE, child.getTcpKeepIdle());
		}
		if (child.getTcpKeepInterval() > 0) {
			boot.option(EpollChannelOption.TCP_KEEPINTVL, child.getTcpKeepInterval());
		}
		if (SocketChannelProvider.SocketType.NATIVE_EPOLL == socketType()) {
			boot.option(EpollChannelOption.TCP_CORK, child.isTcpCork())
					.option(EpollChannelOption.TCP_QUICKACK, child.isTcpQuickAck())
					.option(EpollChannelOption.IP_TRANSPARENT, child.isIpTransparent());
			if (child.isTcpFastOpenConnect()) {
				// Requires Linux kernel 4.11 or later
				boot.option(EpollChannelOption.TCP_FASTOPEN_CONNECT, child.isTcpFastOpenConnect());
			}
			if (child.isEdgeTriggered()) {
				boot.option(EpollChannelOption.EPOLL_MODE, EpollMode.EDGE_TRIGGERED);
			} else {
				boot.option(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
			}
		}
	}

	protected SocketChannelProvider.SocketType socketType() {
		if (Epoll.isAvailable()) {
			return SocketChannelProvider.SocketType.NATIVE_EPOLL;
		}
		if (KQueue.isAvailable()) {
			return SocketChannelProvider.SocketType.NATIVE_KQUEUE;
		}
		return SocketChannelProvider.SocketType.JAVA_NIO;
	}

	@Override
	public Connection connect(UnResolvedAddress address) {
		return connect(address, false);
	}

	@Override
	public TcpConfig config() {
		return childConfig;
	}

	@Override
	protected EventLoopGroup initEventLoopGroup(int nThreads, ThreadFactory tFactory) {
		SocketChannelProvider.SocketType socketType = socketType();
		switch (socketType) {
			case NATIVE_EPOLL:
				return new EpollEventLoopGroup(nThreads, tFactory);
			case NATIVE_KQUEUE:
				return new KQueueEventLoopGroup(nThreads, tFactory);
			case JAVA_NIO:
				return new NioEventLoopGroup(nThreads, tFactory);
			default:
				throw new IllegalStateException("Invalid socket type: " + socketType);
		}
	}

	protected void initChannelFactory() {
		SocketChannelProvider.SocketType socketType = socketType();
		switch (socketType) {
			case NATIVE_EPOLL:
				bootstrap().channelFactory(SocketChannelProvider.NATIVE_EPOLL_CONNECTOR);
				break;
			case NATIVE_KQUEUE:
				bootstrap().channelFactory(SocketChannelProvider.NATIVE_KQUEUE_CONNECTOR);
				break;
			case JAVA_NIO:
				bootstrap().channelFactory(SocketChannelProvider.JAVA_NIO_CONNECTOR);
				break;
			default:
				throw new IllegalStateException("Invalid socket type: " + socketType);
		}
	}
}
